package com.atguigu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.app.BaseAppV1;
import com.atguigu.realtime.bean.TradeUserSpuOrderBean;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.util.DimUtil;
import com.atguigu.realtime.util.JdbcUtil;
import com.atguigu.realtime.util.RedisUtil;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import redis.clients.jedis.Jedis;

import java.sql.Connection;

/**
 * @Author lzc
 * @Date 2022/5/30 10:03
 */
public class Dws_09_DwsTradeUserSpuOrderWindow_Cache extends BaseAppV1 {
    public static void main(String[] args) {
        new Dws_09_DwsTradeUserSpuOrderWindow_Cache().init(3008, 2, "Dws_09_DwsTradeUserSpuOrderWindow_Cache", Constant.TOPIC_DWD_TRADE_ORDER_DETAIL);
    }
    
    @Override
    protected void handle(StreamExecutionEnvironment env,
                          DataStreamSource<String> stream) {
        // 1. 去重: 按照订单详情id去重, 保留最后一条
        SingleOutputStreamOperator<JSONObject> distinctedStream = distinctByOrderDetailId(stream);
        
        // 2. 把数据封装到pojo中
        SingleOutputStreamOperator<TradeUserSpuOrderBean> beanStream = parseToBean(distinctedStream);
        
        // 3. 根据skuId去查找sku_info关联上spu_id和c3_id和tm_id信息. 其他维度信息等到聚合之后再去关联
        SingleOutputStreamOperator<TradeUserSpuOrderBean> beanStreamWithSpu = addSpuIdAndC3IdAndTmId(beanStream);
        beanStreamWithSpu.print();
    }
    
    
    private SingleOutputStreamOperator<TradeUserSpuOrderBean> addSpuIdAndC3IdAndTmId(
        SingleOutputStreamOperator<TradeUserSpuOrderBean> stream) {
        /*
            根据 sku_id去phoenix查找对应的维度信息
        
         */
        return stream
            .map(new RichMapFunction<TradeUserSpuOrderBean, TradeUserSpuOrderBean>() {
    
                private Jedis redisClient;
                private Connection conn;
                
                @Override
                public void open(Configuration parameters) throws Exception {
                    conn = JdbcUtil.getPhoenixConnection(Constant.PHOENIX_DRIVER, Constant.PHOENIX_URL);
    
                    redisClient = RedisUtil.getRedisClient();
                }
                
                @Override
                public void close() throws Exception {
                    if (conn != null && !conn.isClosed()) {
                        conn.close();
                    }
    
                    if (redisClient != null) {
                        redisClient.close();// 如果客户端对象是通过连接池获取的, 则是把连接归还给连接池. 如果是通过new 的方式创建的, 则是关闭客户端
                    }
                }
                
                @Override
                public TradeUserSpuOrderBean map(TradeUserSpuOrderBean bean) throws Exception {
                    
//                    JSONObject obj = DimUtil.getDimFromPhoenix(conn, "dim_sku_info", bean.getSkuId());
                    
                    JSONObject skuInfo = DimUtil.getDim(redisClient, conn, "dim_sku_info", bean.getSkuId());
                    
                    bean.setSpuId(skuInfo.getString("SPU_ID"));// phoenix中所有字段大写
                    bean.setCategory3Id(skuInfo.getString("CATEGORY3_ID"));// phoenix中所有字段大写
                    bean.setTrademarkId(skuInfo.getString("TM_ID"));// phoenix中所有字段大写
                    return bean;
                }
            });
        
    }
    
    
    private SingleOutputStreamOperator<TradeUserSpuOrderBean> parseToBean(
        SingleOutputStreamOperator<JSONObject> stream) {
        return stream
            .map(new MapFunction<JSONObject, TradeUserSpuOrderBean>() {
                
                
                @Override
                public TradeUserSpuOrderBean map(JSONObject value) throws Exception {
                    
                    TradeUserSpuOrderBean bean = new TradeUserSpuOrderBean();
                    bean.setSkuId(value.getString("sku_id"));
                    bean.getOrderIdSet().add(value.getString("order_id"));
                    bean.setUserId(value.getString("user_id"));
                    
                    bean.setOrderAmount(value.getDouble("split_total_amount"));
                    
                    
                    return bean;
                }
            });
    }
    
    
    private SingleOutputStreamOperator<JSONObject> distinctByOrderDetailId(DataStreamSource<String> stream) {
        /*
        去重的优化:
            没有用到右边的数据, 所以只要第一条就可以了
            
         
         */
        return stream
            .map(JSON::parseObject)
            .keyBy(obj -> obj.getString("id"))
            .process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                
                private ValueState<JSONObject> lastOrderDetailState;
                
                @Override
                public void open(Configuration parameters) throws Exception {
                    // 状态中存储数据, 每新来一条数据,就比较时间, 新来的大于状态中的,更新状态中的数据
                    lastOrderDetailState = getRuntimeContext().getState(new ValueStateDescriptor<JSONObject>("lastOrderDetailState", JSONObject.class));
                }
                
                @Override
                public void processElement(JSONObject obj,
                                           Context ctx,
                                           Collector<JSONObject> out) throws Exception {
                    JSONObject lastOrderDetail = lastOrderDetailState.value();
                    if (lastOrderDetail == null) {
                        lastOrderDetailState.update(obj);
                        out.collect(obj);
                    }
                }
            });
        
        
    }
}
/*
缓存优化:

flink 状态
 好处:
    本地内存 读写速度极快
 坏处:
    1. 占用flink的内存, 应该flink的计算
    2. 当维度发生变化的, 状态的数据没有办法更新
 
外置缓存(旁路缓存):
redis
    好处
        当维度发生变化的, redis只的缓存数据可以及时更新
    
    坏处
        1. 要通过网络
        2. 缓存过大, 对内存占用比较大
    
    
    
 */